package bblnamed

import (
	"bblcloud/bblvolume"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/boltdb/bolt"
	"path"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	OFFLINE = iota
	ONLINE
	MAINTENANCE
	REBALANCE
	MALFUNCTION
)

const (
	VOLUME_TIMEOUT = 60
)

var (
	cluster_db = "cluster.db"
	job_bucket = []byte("jobs")
	//volume_nodes_bucket = []byte("volume_nodes")
	//client_nodes_bucket   = []byte("client_nodes")
	cluster_config_bucket = []byte("server_config")
)

func NewClusterManager(workdir string) (ClusterManager, error) {
	p := new(manager)
	p._lock = new(sync.Mutex)
	p.stat = &ClusterStat{
		status:  ONLINE, //"0:Offline,1:Online,2:Maintenance(readonly),3:Rebalance"
		version: Rand(),
		manager: p,
	}
	db, err := bolt.Open(path.Join(workdir, cluster_db), 0600, &bolt.Options{Timeout: 30 * time.Second})
	if err != nil {
		return nil, err
	}
	p.db = db
	p.elastichash = NewElasticHash()
	err = p.db.Update(func(tx *bolt.Tx) error {
		_, err := tx.CreateBucketIfNotExists(job_bucket)
		if err != nil {
			return err
		}
		// _, err = tx.CreateBucketIfNotExists(volume_nodes_bucket)
		// if err != nil {
		// 	return err
		// }
		_, err = tx.CreateBucketIfNotExists(cluster_config_bucket)
		if err != nil {
			return err
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	tx, err := p.db.Begin(true)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	// Use the transaction...
	bucket := tx.Bucket(cluster_config_bucket)

	if val := bucket.Get([]byte("cluster.uuid")); val != nil {
		p.uuid = string(val)
	} else {
		p.uuid = server_conf.UUID
		err = bucket.Put([]byte("cluster.uuid"), []byte(p.uuid))
		if err != nil {
			return nil, err
		}
	}
	if val := bucket.Get([]byte("cluster.name")); val != nil {
		p.name = string(val)
	} else {
		p.name = server_conf.Name
		err = bucket.Put([]byte("cluster.name"), []byte(p.name))
		if err != nil {
			return nil, err
		}
	}
	if val := bucket.Get([]byte("cluster.replicas")); val != nil {
		i, _ := strconv.Atoi(string(val))
		p.replicas = i
	} else {
		p.replicas = server_conf.Replicas
		err = bucket.Put([]byte("cluster.replicas"), []byte(fmt.Sprintf("%v", p.replicas)))
		if err != nil {
			return nil, err
		}
	}
	if val := bucket.Get([]byte("cluster.nodes")); val != nil {
		json.Unmarshal(val, &p.nodes)
	} else {
		p.nodes = make(map[string]*Volume)
	}
	if val := bucket.Get([]byte("cluster.clients")); val != nil {
		json.Unmarshal(val, &p.clients)
	} else {
		p.clients = make(map[string]*Gateway)
	}
	if val := bucket.Get([]byte("cluster.stat.totalSpace")); val != nil {
		i, _ := strconv.Atoi(string(val))
		p.stat.totalSpace = int64(i)
	}
	if val := bucket.Get([]byte("cluster.stat.usedSpace")); val != nil {
		i, _ := strconv.Atoi(string(val))
		p.stat.usedSpace = int64(i)
	}
	if val := bucket.Get([]byte("cluster.stat.status")); val != nil {
		i, _ := strconv.Atoi(string(val))
		p.stat.status = i
	} else {
		err = bucket.Put([]byte("cluster.stat.status"), []byte(fmt.Sprintf("%v", p.stat.status)))
		if err != nil {
			return nil, err
		}
	}
	if val := bucket.Get([]byte("cluster.stat.version")); val != nil {
		i, _ := strconv.Atoi(string(val))
		p.stat.version = i
	} else {
		err = bucket.Put([]byte("cluster.stat.version"), []byte(fmt.Sprintf("%v", p.stat.version)))
		if err != nil {
			return nil, err
		}
	}
	for addr, _ := range p.clients {
		if _, err = tx.CreateBucketIfNotExists([]byte("gw_job_" + addr)); err != nil {
			return nil, err
		}
	}
	for uuid, _ := range p.nodes {
		if _, err = tx.CreateBucketIfNotExists([]byte("vol_job_" + uuid)); err != nil {
			return nil, err
		}
		p.elastichash.Add(uuid)
	}

	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return p, nil
}

type ClusterManager interface {
	UUID() string
	Name() string
	AddVolume(uuid string) error
	RemoveVolume(uuid string) error
	AddGateway(addr, perm string) error
	RemoveGateway(addr string) error
	ValidGateway(addr string) bool
	ListGateway() ([]*Gateway, error)
	JoinVolume(node *Volume) error
	LeaveVolume(uuid string) error
	FindOneVolume(uuid string) (*Volume, error)
	FindVolume(uuid []string) ([]*Volume, error)
	LocateVolume(path string) ([]*Volume, error)
	ListVolume() ([]*Volume, error)
	ClusterStat() *ClusterStat
	LookupJob(uuid string, size int) ([]*Job, error)
	PutJob(uuid string, job *Job) error
	RemoveJob(uuid, jobid string) error
	RemoveAllJob(uuid string) error
	Close() error
}
type Job struct {
	Id    string
	Name  string
	CTime time.Time
	Args  map[string]string
}

func newJob(name string) *Job {
	return &Job{Name: name, CTime: time.Now(), Args: make(map[string]string)}
}
func (j *Job) encode() ([]byte, error) {
	return json.Marshal(j)
}

func (j *Job) decode(b []byte) error {
	return json.Unmarshal(b, j)
}

type Volume struct {
	UUID    string //"550e8400-e29b-41d4-a716-446655440000",
	Name    string //Storage#1
	Address string //"192.168.1.1:2003",
	// TotalSpace         uint64
	// UsedSpace          uint64
	Export string //"/export/storage",
	// MountDev           string //"/dev/md1"
	Status             int //"0:Offline,1:Online,2:Maintenance,3:Rebalance"
	HeartbeatTime      time.Time
	MalfunctionNum     int
	LastMalfunctionNum time.Time
	Stat               *bblvolume.VolStat
}

func (p *Volume) String() string {
	b, err := json.Marshal(*p)
	if err != nil {
		return fmt.Sprintf("%v", *p)
	}
	return string(b)
}
func (p *Volume) CheckHeartbeat() {
	now := time.Now().Unix()
	if now-p.HeartbeatTime.Unix() > VOLUME_TIMEOUT {
		p.Status = OFFLINE
	}
}

type Gateway struct {
	Address string
	Perm    string //rwd
}

func (p *Gateway) String() string {
	b, err := json.Marshal(*p)
	if err != nil {
		return fmt.Sprintf("%v", *p)
	}
	return string(b)
}

type ClusterStat struct {
	totalSpace int64
	usedSpace  int64
	status     int //"0:Offline,1:Online,2:Maintenance(readonly),3:Rebalance"
	version    int
	manager    *manager
}

func (c *ClusterStat) TotalSpace() int64 {
	return c.totalSpace
}
func (c *ClusterStat) UsedSpace() int64 {
	return c.usedSpace
}
func (c *ClusterStat) Status() int {
	return c.status
}
func (c *ClusterStat) Version() int {
	return c.version
}
func (c *ClusterStat) Volumes() []*Volume {
	nodes, _ := c.manager.ListVolume()
	return nodes
}

func (c *ClusterStat) Clients() []*Gateway {
	clients, _ := c.manager.ListGateway()
	return clients
}

var ClosedError = errors.New("cluster mananger is closed")

type manager struct {
	uuid        string
	name        string
	replicas    int //default 3
	nodes       map[string]*Volume
	clients     map[string]*Gateway
	_lock       *sync.Mutex
	stat        *ClusterStat
	db          *bolt.DB
	closed      bool
	elastichash *Consistent
}

func (p *manager) UUID() string {
	return p.uuid
}
func (p *manager) Name() string {
	return p.name
}

func (p *manager) AddGateway(addr string, perm string) error {
	if p.is_closed() {
		return ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	var c *Gateway
	if gw, exist := p.clients[addr]; exist {
		log.Debug(addr, perm, gw)
		gw.Perm = perm
		c = gw
	} else {
		c = &Gateway{Address: addr, Perm: perm}
	}
	log.Debug(addr, perm)
	p.clients[addr] = c
	if err := p.db.Update(func(tx *bolt.Tx) error {
		data, err := json.Marshal(p.clients)
		if err != nil {
			return err
		}
		if err = tx.Bucket(cluster_config_bucket).Put([]byte("cluster.clients"), data); err != nil {
			return err
		}
		//创建gw任务表
		if _, err = tx.CreateBucketIfNotExists([]byte("gw_job_" + addr)); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return err
	}
	return nil
}
func (p *manager) RemoveGateway(addr string) error {
	if p.is_closed() {
		return ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	if _, exist := p.clients[addr]; !exist {
		return fmt.Errorf("Gateway %v Not Exist", addr)
	}
	delete(p.clients, addr)
	if err := p.db.Update(func(tx *bolt.Tx) error {
		data, err := json.Marshal(p.clients)
		if err != nil {
			return err
		}
		if err = tx.Bucket(cluster_config_bucket).Put([]byte("cluster.clients"), data); err != nil {
			return err
		}
		if err = tx.DeleteBucket([]byte("gw_job_" + addr)); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return err
	}
	return nil
}

func (p *manager) ValidGateway(addr string) bool {
	if p.is_closed() {
		return false
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	for key, _ := range p.clients {
		if strings.HasPrefix(key, addr) {
			return true
		}
	}
	for _, val := range p.nodes {
		if strings.HasPrefix(val.Address, addr) {
			return true
		}
	}
	return false
}

func (p *manager) ListGateway() ([]*Gateway, error) {
	if p.is_closed() {
		return nil, ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	clients := make([]*Gateway, 0)
	for _, client := range p.clients {
		clients = append(clients, client)
	}
	return clients, nil
}

func (p *manager) AddVolume(uuid string) error {
	if p.is_closed() {
		return ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	if _, exist := p.nodes[uuid]; exist {
		return fmt.Errorf("Volume %v Exist", uuid)
	}
	vol := &Volume{UUID: uuid}
	p.nodes[uuid] = vol
	if err := p.db.Update(func(tx *bolt.Tx) error {
		data, err := json.Marshal(p.nodes)
		if err != nil {
			return err
		}
		if err = tx.Bucket(cluster_config_bucket).Put([]byte("cluster.nodes"), data); err != nil {
			return err
		}
		if _, err = tx.CreateBucketIfNotExists([]byte("vol_job_" + uuid)); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return err
	}

	p.elastichash.Add(uuid)
	return nil
}
func (p *manager) RemoveVolume(uuid string) error {
	if p.is_closed() {
		return ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	if _, exist := p.nodes[uuid]; !exist {
		return fmt.Errorf("Volume %v Not Exist", uuid)
	}
	delete(p.nodes, uuid)
	p.elastichash.Remove(uuid)
	if err := p.db.Update(func(tx *bolt.Tx) error {
		data, err := json.Marshal(p.nodes)
		if err != nil {
			return err
		}
		if err = tx.Bucket(cluster_config_bucket).Put([]byte("cluster.nodes"), data); err != nil {
			return err
		}
		if err = tx.DeleteBucket([]byte("vol_job_" + uuid)); err != nil {
			return err
		}
		return nil
	}); err != nil {
		return err
	}
	return nil
}

func (p *manager) FindOneVolume(uuid string) (*Volume, error) {
	if p.is_closed() {
		return nil, ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	if n, exist := p.nodes[uuid]; exist {
		n.CheckHeartbeat()
		return n, nil
	}
	return nil, nil
}

func (p *manager) FindVolume(uuids []string) ([]*Volume, error) {
	if p.is_closed() {
		return nil, ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()

	vs := make([]*Volume, 0)
	for _, uuid := range uuids {
		if n, exist := p.nodes[uuid]; exist {
			n.CheckHeartbeat()
			vs = append(vs, n)
		}
	}
	return vs, nil
}

//根据path定位柱服务器
func (p *manager) LocateVolume(path string) ([]*Volume, error) {
	if p.is_closed() {
		return nil, ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	log.Debug("replicas", p.replicas)
	if keys, err := p.elastichash.GetN(path, p.replicas); err != nil {
		return nil, err
	} else {
		log.Debug("elastichash", keys)
		vs := make([]*Volume, 0)
		for _, uuid := range keys {
			if n, exist := p.nodes[uuid]; exist {
				n.CheckHeartbeat()
				vs = append(vs, n)
			}
		}
		return vs, nil
	}
}

func (p *manager) ListVolume() ([]*Volume, error) {
	if p.is_closed() {
		return nil, ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	nodes := make([]*Volume, 0)
	for _, node := range p.nodes {
		node.CheckHeartbeat()
		nodes = append(nodes, node)
	}
	return nodes, nil
}
func (p *manager) JoinVolume(node *Volume) error {
	if p.is_closed() {
		return ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	if n, exist := p.nodes[node.UUID]; exist {
		n.Name = node.Name //Storage#1
		n.Address = node.Address
		// n.TotalSpace = node.TotalSpace
		// n.UsedSpace = node.UsedSpace
		n.Export = node.Export //"/export/storage",
		// n.MountDev = node.MountDev //"/dev/md1"
		n.Status = ONLINE
		n.HeartbeatTime = time.Now()
		log.Infof("node join %s:%s", node.UUID, node.Address)
		return p.db.Update(func(tx *bolt.Tx) error {
			data, err := json.Marshal(p.nodes)
			if err != nil {
				return err
			}
			return tx.Bucket(cluster_config_bucket).Put([]byte("cluster.nodes"), data)
		})
	}
	log.Infof("node join %s:%s %s", node.UUID, node.Address, "Refuse")
	return errors.New("Volume Node " + node.UUID + " Not Exist")
}

func (p *manager) LeaveVolume(uuid string) error {
	if p.is_closed() {
		return ClosedError
	}
	p._lock.Lock()
	defer p._lock.Unlock()
	if n, exist := p.nodes[uuid]; exist {
		n.Status = OFFLINE
		log.Infof("volume leave %s:%s", uuid, n.Address)
		return p.db.Update(func(tx *bolt.Tx) error {
			data, err := json.Marshal(p.nodes)
			if err != nil {
				return err
			}
			return tx.Bucket(cluster_config_bucket).Put([]byte("cluster.nodes"), data)
		})
	}
	return errors.New("Volume Node " + uuid + " Not Exist")
}

func (p *manager) ClusterStat() *ClusterStat {
	p._lock.Lock()
	defer p._lock.Unlock()
	return p.stat
}

func (w *manager) LookupJob(uuid string, size int) (jobs []*Job, err error) {
	if w.is_closed() {
		return nil, ClosedError
	}
	tx, err := w.db.Begin(false)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	jobs = make([]*Job, 0)
	i := 0
	uuid = "vol_job_" + uuid
	err = tx.Bucket([]byte(uuid)).ForEach(func(k, v []byte) error {
		if i >= size {
			return nil
		}
		job := new(Job)
		err := job.decode(v)
		if err != nil {
			return err
		}
		jobs = append(jobs, job)
		i += 1
		return nil
	})
	return jobs, err
}

func (w *manager) PutJob(uuid string, job *Job) error {
	if w.is_closed() {
		return ClosedError
	}
	tx, err := w.db.Begin(true)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	// Use the transaction...
	uuid = "vol_job_" + uuid
	seq, err := tx.Bucket([]byte(uuid)).NextSequence()
	if err != nil {
		return err
	}
	job.Id = fmt.Sprintf("%v", seq)
	val, err := job.encode()
	if err != nil {
		return err
	}
	uuid = "vol_job_" + uuid
	if err = tx.Bucket([]byte(uuid)).Put([]byte(job.Id), val); err != nil {
		return err
	}

	if err := tx.Commit(); err != nil {
		return err
	}
	return nil
}

func (m *manager) RemoveJob(uuid, id string) error {
	if m.is_closed() {
		return ClosedError
	}
	tx, err := m.db.Begin(true)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	uuid = "vol_job_" + uuid
	if err = tx.Bucket([]byte(uuid)).Delete([]byte(id)); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	return nil
}

func (m *manager) RemoveAllJob(uuid string) error {
	if m.is_closed() {
		return ClosedError
	}
	// tx, err := m.db.Begin(true)
	// if err != nil {
	// 	return err
	// }
	// defer tx.Rollback()

	// err = tx.DeleteBucket([]byte(uuid))
	// if err != nil {
	// 	return err
	// }
	// if err := tx.Commit(); err != nil {
	// 	return err
	// }
	return nil
}
func (w *manager) is_closed() bool {
	w._lock.Lock()
	defer w._lock.Unlock()
	return w.closed
}
func (w *manager) Close() error {
	w._lock.Lock()
	defer w._lock.Unlock()
	if w.closed {
		return nil
	}
	w.closed = true
	return w.db.Close()
}
